
#include "include/thread_pool.h"

#include <algorithm>
#include <iostream>

#include "include/task_dispatcher.h"

namespace dispatcher
{

Thread::~Thread()
{
    task_       = nullptr;
    is_started_ = false;
    conditions_.notify_one();
    std::cout << "[ \e[1;35mDEBUG\e[0m ]: Thread Exit, thread_id=" << worker_thread_.get_id() << std::endl;
    worker_thread_.join();
}

void Thread::AddTask(TaskSPtr task_sptr)
{
    if (nullptr == task_sptr)
    {
        return;
    }

    {
        std::lock_guard<std::mutex> locker(wait_task_lock_);
        task_ = task_sptr;
    }
    conditions_.notify_one();
}

void Thread::Run()
{
    is_started_ = true;
    for (; is_started_;)
    {
        std::unique_lock<std::mutex> locker(wait_task_lock_);
        conditions_.wait(locker);

        if (nullptr == task_)
        {
            break;
        }

        task_->Run();
        task_->SetState(Task::Finished);

        finished_callback_(shared_from_this());
    }
}

namespace thread_pool
{

ThreadPool::ThisType ThreadPool::s_thread_pool_ = nullptr;

ThreadPool::ThisType ThreadPool::GetInstance()
{
    if (nullptr == s_thread_pool_)
    {
        s_thread_pool_ = std::shared_ptr<ThreadPool>(new ThreadPool);
    }
    return s_thread_pool_;
}

void ThreadPool::WorkerThreadReleased(WorkerThread worker)
{
    auto iter = std::find(busy_workers_.begin(), busy_workers_.end(), worker);
    if (iter != busy_workers_.end())
    {
        busy_workers_.erase(iter);
        idle_workers_.emplace_back(worker);
    }
    else
    {
        std::cerr << "\e[1;31mReleased worker thread not in busy_workers.\e[0m" << std::endl;
    }
}

ThreadPool::WorkerThread ThreadPool::CreateWorker()
{
    return std::make_shared<Thread>(std::bind(&ThreadPool::WorkerThreadReleased, this, std::placeholders::_1));
}

bool ThreadPool::Start(TaskSPtr task)
{
    if (!IsAvaliable())
    {
        return false;
    }

    WorkerThread worker;

    if (idle_workers_.size())
    {
        // Use IDLE worker thread.
        worker = idle_workers_.front();
        idle_workers_.pop_front();
    }
    else
    {
        // Create new worker thread.
        worker = CreateWorker();
    }

    worker->AddTask(task);
    busy_workers_.emplace_back(worker);
    return true;
}

}  // namespace thread_pool

}  // namespace dispatcher
